package com.gitee.wsl.coroutines.executor.dispatcher

import com.gitee.wsl.coroutines.executor.sample.AsyncExecutor
import com.gitee.wsl.ext.base.currentTime
import com.gitee.wsl.ext.coroutines.launchUnScoped
import kotlinx.atomicfu.locks.SynchronizedObject
import kotlinx.atomicfu.locks.synchronized
import kotlinx.coroutines.CancellableContinuation
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.DisposableHandle
import kotlinx.coroutines.Runnable
import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException
import kotlin.time.Duration
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.ExperimentalTime
import kotlin.time.Instant


/**
 * A [CoroutineDispatcher] that wraps around an [AsyncExecutor] instance ([executor]) to execute
 * tasks asynchronously.
 *
 * Requires calling [executePending] in order to support [delay].
 */
class AsyncThreadDispatcher(
    val executor: AsyncExecutor = AsyncExecutor.INSTANCE,
    val threads: Int = -1,
) : CoroutineDispatcher(),KtDispatcher {
    private val lock: SynchronizedObject = SynchronizedObject()

    private val tasks = mutableListOf<Runnable>()
    private val timedTasks: MutableList<TimedTask> = mutableListOf()

    override fun dispatch(context: CoroutineContext, block: Runnable) {
          execute(block)
    }

    override fun execute(block: Runnable) = launchUnScoped {
        executor.executeSuspend{ block.run() }
    }

    override fun queue(block: Runnable) {
        synchronized(lock) { tasks += block }
    }

    suspend fun close() {
        executor.cancel()
    }

   @OptIn(ExperimentalTime::class)
   override fun invokeOnTimeout(
        timeMillis: Long,
        block: Runnable,
        context: CoroutineContext
    ): DisposableHandle {
        val task = TimedTask(currentTime + timeMillis.milliseconds, null, block)
         synchronized(lock) { timedTasks += task }
        return DisposableHandle { synchronized(lock) { timedTasks -= task } }
    }

     override fun scheduleResumeAfterDelay(
        timeMillis: Long,
        continuation: CancellableContinuation<Unit>
    ) {
        scheduleResumeAfterDelay(timeMillis.milliseconds, continuation)
    }

    /** Executes any pending timed ([delay]) or queued ([queue]) tasks. */
    @OptIn(ExperimentalTime::class)
    fun executePending() {
        try {
            //val startTime =
            executeTimedTasks(currentTime)
            executeQueuedTasks()
        } catch (e: Throwable) {
            e.printStackTrace()
        }
    }

    @OptIn(ExperimentalTime::class)
    private fun scheduleResumeAfterDelay(
        time: Duration,
        continuation: CancellableContinuation<Unit>
    ) {
        val task = TimedTask(currentTime + time, continuation, null)
        continuation.invokeOnCancellation { task.exception = it }
        synchronized(lock) { timedTasks += task }
    }

    @OptIn(ExperimentalTime::class)
    private fun executeTimedTasks(startTime: Instant) {
        while (true) {
            val item =
                synchronized(lock) {
                    if (timedTasks.isNotEmpty() && startTime >= timedTasks.first().time) {
                        timedTasks.removeFirst()
                    } else {
                        null
                    }
                } ?: break
            item.exception?.let { exception ->
                item.continuation?.resumeWithException(exception)
                item.callback?.let { exception.printStackTrace() }
            }
                ?: run {
                    item.continuation?.resume(Unit)
                    item.callback?.run()
                }
        }
    }

    private fun executeQueuedTasks() {
        while (true) {
            val task =
                synchronized(lock) {
                    if (tasks.isNotEmpty()) {
                        tasks.removeFirst()
                    } else {
                        null
                    }
                } ?: break
            task.run()
        }
    }

    override fun toString(): String {
        return "AsyncThreadDispatcher(executor=$executor, threads=$threads)"
    }

    private class TimedTask @OptIn(ExperimentalTime::class) constructor(
        val time: Instant,
        val continuation: CancellableContinuation<Unit>?,
        val callback: Runnable?,
    ) {
        var exception: Throwable? = null
    }

    companion object{
        val DEFAULT =  AsyncThreadDispatcher()
    }
}

val Dispatchers.asyncThread get() = AsyncThreadDispatcher.DEFAULT